package com.atguigu.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * 使用canal获取mysql中inset变化后的数据发送到kafka
 *
 * @author WangJX
 * @version 1.0
 * @date 2019/11/28 9:58
 */
public class CanalApp {
    public static void main(String[] args) {

        //建立canal的连接，canal没有设置账号和密码
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("hadoop102", 11111),
                "example",  //指定数据源
                "",
                "");

        //建立高可用的canal连接
        //CanalConnector connector2 = CanalConnectors.newClusterConnector("","example", "", "");


        while (true) {
            //链接对应的canal server
            connector.connect();

            //订阅所有库     订阅指定库*.gmall_parent
            connector.subscribe("*.*");

            //一次获取100条sql的数据
            Message message = connector.get(100);

            int size = message.getEntries().size();
            //判断是否抓取到数据
            if (size <= 0) {
                try {
                    System.out.println("没有数据，休息5秒");
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                //循环遍历一条sql所变化的数据
                for (CanalEntry.Entry entry : message.getEntries()) {
                    //获取具体的所有数据(经过序列化后的数据)
                    ByteString storeValue = entry.getStoreValue();

                    CanalEntry.RowChange rowChange = null;
                    try {
                        //使用CanalEntry.RowChange工具进行反序列化
                        rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                    } catch (InvalidProtocolBufferException e) {
                        e.printStackTrace();
                    }

                    //获取行级数据
                    List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                    //获取表名
                    String tableName = entry.getHeader().getTableName();
                    //获取操作类型 insert  update delete
                    CanalEntry.EventType eventType = rowChange.getEventType();

                    //封装数据
                    CanalHandler canalHandler = new CanalHandler(tableName, eventType, rowDatasList);
                    //解析并发送数据到kafka
                    canalHandler.handle();
                }
            }
        }


    }
}
